package com.shujia.utils;

import com.shujia.bean.MyCell;
import com.shujia.bean.NotFoundFilterNameException;
import com.shujia.bean.NotFountComparatorException;
import jdk.nashorn.internal.codegen.types.BooleanType;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.*;

public class HBaseUtil {
    public static Connection CONNECTION;
    public static Admin ADMIN;

    private HBaseUtil() {
    }

    /**
     * 获取数据库连接对象和数据库操作对象
     */
    static {
        try {
            //1、获取hbase集群的配置文件对象
            //0.90.0之前旧版本的写法：
//        HBaseConfiguration conf = new HBaseConfiguration();
            //新版本的写法：调用静态方法public static Configuration create()
            Configuration conf = HBaseConfiguration.create();

            //2、因为hbase的数据都有一条元数据，而元数据也存储在一张表中，这张元数据表也有元数据，存储在zookeeper中
            //配置文件要设置你自己的zookeeper集群
//            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181"); //前提时windows中配置了hosts映射
            Properties prop = new Properties();
            prop.load(new BufferedReader(new FileReader("E:\\projects\\IDEAProjects\\bigdata29\\hbase\\src\\main\\resources\\hbase.properties")));

            String zk = (String) prop.get("hbase.zookeeper.quorum");

            //2、因为hbase的数据都有一条元数据，而元数据也存储在一张表中，这张元数据表也有元数据，存储在zookeeper中
            //配置文件要设置你自己的zookeeper集群
            conf.set("hbase.zookeeper.quorum", zk); //前提时windows中配置了hosts映射

            //3、获取数据库的连接对象
            CONNECTION = ConnectionFactory.createConnection(conf);

            //4、获取数据库操作对象
//            HBaseAdmin hBaseAdmin = new HBaseAdmin(conn);
            //新版本的做法
            ADMIN = CONNECTION.getAdmin(); //使用连接对象获取数据库操作对象
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void createOneTable(String tableName, String columnFamily) {
        try {
            //先将表名封装成一个TableName的对象
            TableName tn = TableName.valueOf(tableName);
            if (!ADMIN.tableExists(tn)) {
                TableDescriptorBuilder students = TableDescriptorBuilder.newBuilder(tn);
//                ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of(columnFamily);

                //使用另外一种方式创建列簇并设置布隆过滤器
                ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily))
                        .setBloomFilterType(BloomType.ROW).build();


                students.setColumnFamily(columnFamilyDescriptor);
                ADMIN.createTable(students.build());
                System.out.println(tn + "表创建成功！！！");
            } else {
                System.out.println(tn + "表已经存在！");
            }
        } catch (Exception e) {
            System.out.println("表创建失败！！");
            e.printStackTrace();
        }
    }

    /**
     * 需求：2、如何删除一张表
     * disable 'students'
     * drop 'students'
     */
    public static void dropOneTable(String tableName) {
        try {
            //先将表名封装成一个TableName的对象
            TableName tn = TableName.valueOf(tableName);
            //判断表是否存在
            if (ADMIN.tableExists(tn)) {
                //先禁用表
                ADMIN.disableTable(tn);
                ADMIN.deleteTable(tn);
                System.out.println(tn + "表成功被删除");
            } else {
                System.out.println(tn + "表不存在！！");
            }


        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 需求：3、如何向一张表中添加一条数据
     * put 'students','1001','info:name','小虎'
     */
    public static void putOneDataToTable(String tableName, String rowKey, String columnFamily, String colName, String colValue) {
        try {
            //先将表名封装成一个TableName的对象
            TableName tn = TableName.valueOf(tableName);

            //判断表是否存在
            if (ADMIN.tableExists(tn)) {
                //获取表对象
                Table students = CONNECTION.getTable(tn);
                //hbase自带的一个工具类Bytes，可以将字符串转字节数组
                Put put = new Put(Bytes.toBytes(rowKey));//行键的字节数组形式
                put.add(new KeyValue(Bytes.toBytes(rowKey),
                        Bytes.toBytes(columnFamily),
                        Bytes.toBytes(colName),
                        Bytes.toBytes(colValue)));
                //void put(Put put)
                //需要先将我们添加的列数据封装成一个Put对象
                students.put(put);
            } else {
                System.out.println(tn + "表不存在！！");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 需求：5、如何获取一条数据
     * get 'students','1500101000'
     */
    public static List<MyCell> getOneData(String tableName, String rowKey) throws Exception {
        //将表名封装TableName对象
        TableName sd = TableName.valueOf(tableName);
        //获取表的对象
        Table students = CONNECTION.getTable(sd);

        //传入行键的字节数组的形式
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = students.get(get);

        //创建List集合List<MyCell>  MyCell:行键，列簇，列名，列值，
        ArrayList<MyCell> myCells = new ArrayList<>();
        byte[] rk = result.getRow();
        //当不清楚列名的时候该如何获取呢？
        //获取一行中的所有单元格组合的集合
        List<Cell> cells = result.listCells(); //获取的结果值的顺序是hbase中存储排序后的顺序
        //遍历集合得到每个单元格，获取每个列值
        for (Cell cell : cells) {
            byte[] family = CellUtil.cloneFamily(cell);
            byte[] qualifier = CellUtil.cloneQualifier(cell);
            byte[] value = CellUtil.cloneValue(cell);
            myCells.add(new MyCell(rk, family, qualifier, value));
        }
        return myCells;
    }


    /**
     * 7、如果获取一批数据 第二种方式
     * scan 'students' 默认情况下获取表所有数据
     * scan 'students',LIMIT=>3
     */
    public static void scanData(String tableName) {
        try {
            TableName sd = TableName.valueOf(tableName);
            Table students = CONNECTION.getTable(sd);
            //创建Scan对象
            Scan scan = new Scan(); //查询表中的所有行

            //default ResultScanner getScanner(Scan scan)
            ResultScanner resultScanner = students.getScanner(scan);
            //通过观察源码发现，可以先获取迭代器对象
            for (Result result : resultScanner) {
                printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void scanData(String tableName, String startRow, String endRow, boolean include) {
        try {
            TableName sd = TableName.valueOf(tableName);
            Table students = CONNECTION.getTable(sd);
            //创建Scan对象
            Scan scan = new Scan(); //查询表中的所有行
            //设置开始行和结束行
            if (include) {
                scan.withStartRow(Bytes.toBytes(startRow));
                scan.withStopRow(Bytes.toBytes(endRow), true);
            } else {
                scan.withStartRow(Bytes.toBytes(startRow));
                scan.withStopRow(Bytes.toBytes(endRow));
            }

            //default ResultScanner getScanner(Scan scan)
            ResultScanner resultScanner = students.getScanner(scan);
            //通过观察源码发现，可以先获取迭代器对象
            for (Result result : resultScanner) {
                printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void scanData(String tableName, int limit) {
        try {
            TableName sd = TableName.valueOf(tableName);
            Table students = CONNECTION.getTable(sd);
            //创建Scan对象
            Scan scan = new Scan(); //查询表中的所有行
            scan.setLimit(limit);

            //default ResultScanner getScanner(Scan scan)
            ResultScanner resultScanner = students.getScanner(scan);
            //通过观察源码发现，可以先获取迭代器对象
            for (Result result : resultScanner) {
                printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static ResultScanner scanDataReturn(String tableName, int limit) throws Exception{
        TableName sd = TableName.valueOf(tableName);
        Table students = CONNECTION.getTable(sd);
        //创建Scan对象
        Scan scan = new Scan(); //查询表中的所有行
        scan.setLimit(limit);

        //default ResultScanner getScanner(Scan scan)
        return students.getScanner(scan);
    }

    /**
     * 需求：7、如何创建预分region表
     * 将来对于某一个RegionServer而言，可能会接收大量的并发请求，超出了该RegionServer承受的压力，有可能会导致该子节点崩溃
     * 如果大量的并发请求查询的数据是多种多种多样的，只不过巧合的是都在一个RegionServer上管理
     * 解决的思路：分散查询的数据到不同的RegionServer上，这样请求也会随着被分散到不同的RegionServer上，就达到了减轻某一个RegionServer压力过大情况，解决了单点故障的问题
     */
    public static void createPreviewTable(String tableName, String columnFamily, byte[][] splitKeys) {
        try {
            //先将表名封装成TableName对象
            TableName tb = TableName.valueOf(tableName);

            //创建表描述器对象
            TableDescriptorBuilder tbSplit2 = TableDescriptorBuilder.newBuilder(tb);

            //创建列簇描述器对象
            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of(columnFamily);

            //将列簇添加到表中
            tbSplit2.setColumnFamily(info);


            //调用方法创建表
//            admin.createTable(tbSplit2.build());
            //调用另外一个方法，传入表描述器的同时，传入分割点，创建预分region表
            //void createTable(TableDescriptor desc, byte[][] splitKeys)
            ADMIN.createTable(tbSplit2.build(), splitKeys);
            System.out.println(tableName + "预分region表创建成功！！！");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 编写一个方法专门做比较过滤器的
     * <p>
     * 需要考虑的参数：表名 过滤器 操作符 比较器 值
     */
    public static void scanDataWithCompareFilter(String tableName, String filterType, CompareOperator compareOperator,
                                                 String comparatorType, String value) {
        try {
            /**
             * CompareFilter 是一个抽象类，无法被实例化
             * 但是，我们通过观察源码，可以得到一些信息，创建比较过滤器对象的时候，至少需要传入两种参数：操作符，比较器
             */
            //获取表对象
            TableName st = TableName.valueOf(tableName);
            Table students = CONNECTION.getTable(st);
            Scan scan = new Scan();
            ByteArrayComparable byteArrayComparable = null;

            switch (comparatorType) {
                case "BinaryComparator":
                    byteArrayComparable = new BinaryComparator(Bytes.toBytes(value));
                    break;
                case "BinaryPrefixComparator":
                    byteArrayComparable = new BinaryPrefixComparator(Bytes.toBytes(value));
                    break;
                case "SubstringComparator":
                    byteArrayComparable = new SubstringComparator(value);
                    break;
                default:
                    throw new NotFountComparatorException("没有匹配的比较器！使用ComparableName获取");
            }


            //--------------------------------------------------------------------------------
            CompareFilter compareFilter = null;
            switch (filterType) {
                case "RowFilter":
                    compareFilter = new RowFilter(compareOperator, byteArrayComparable);
                    break;
                case "FamilyFilter":
                    compareFilter = new FamilyFilter(compareOperator, byteArrayComparable);
                    break;
                case "QualifierFilter":
                    compareFilter = new QualifierFilter(compareOperator, byteArrayComparable);
                    break;
                case "ValueFilter":
                    compareFilter = new ValueFilter(compareOperator, byteArrayComparable);
                    break;
                default:
                    throw new NotFoundFilterNameException("没有匹配的比较过滤器！使用FilterName获取");
            }

            //--------------------------------------------------------------------------------

            //设置过滤器
            //public Scan setFilter(Filter filter)
            scan.setFilter(compareFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            for (Result result : resultScanner) {
                HBaseUtil.printResult(result);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void printResult(Result result) {
        List<Cell> cells = result.listCells();
        StringBuffer sb = new StringBuffer();

        String rk = Bytes.toString(result.getRow());
        sb.append(rk).append(",");
        for (int i = 0; i < cells.size(); i++) {
            String fa = Bytes.toString(CellUtil.cloneFamily(cells.get(i)));
            String qName = Bytes.toString(CellUtil.cloneQualifier(cells.get(i)));
            String colName = Bytes.toString(CellUtil.cloneValue(cells.get(i)));
            if (i == cells.size() - 1) {
                sb.append(fa).append(":").append(qName).append("-").append(colName); // 1001,info:name-xiaohu,info:age-18
            } else {
                sb.append(fa).append(":").append(qName).append("-").append(colName).append(",");
            }
        }
        System.out.println(sb);
    }


    /**
     * 释放Before创建的资源，在每个Test之后执行
     */
    public static void closeSource() {
        try {
            if (ADMIN != null) {
                ADMIN.close();
            }

            if (CONNECTION != null) {
                CONNECTION.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
